// This file is licensed under the Elastic License 2.0. Copyright 2021 StarRocks Limited.

#pragma once

#include "common/status.h"
#include "exec/olap_common.h"
#include "exprs/expr.h"
#include "exprs/expr_context.h"

namespace starrocks {
namespace vectorized {

class RuntimeFilterProbeCollector;
class PredicateParser;
class ColumnPredicate;

class OlapScanConjunctsManager {
public:
    // fields from olap scan node
    const std::vector<ExprContext*>* conjunct_ctxs_ptr;
    const TupleDescriptor* tuple_desc;
    ObjectPool* obj_pool;
    const std::vector<std::string>* key_column_names;
    const RuntimeFilterProbeCollector* runtime_filters;

private:
    // fields generated by parsing conjunct ctxs.
    // same size with |_conjunct_ctxs|, indicate which element has been normalized.
    std::vector<bool> normalized_conjuncts;
    std::map<std::string, ColumnValueRangeType> column_value_ranges; // from conjunct_ctxs
    OlapScanKeys scan_keys;                                          // from _column_value_ranges
    std::vector<TCondition> olap_filters;                            // from _column_value_ranges
    std::vector<TCondition> is_null_vector;                          // from conjunct_ctxs

public:
    static void eval_const_conjuncts(const std::vector<ExprContext*>& conjunct_ctxs, Status* status);

    void parse_to_column_predicates(PredicateParser* parser, std::vector<ColumnPredicate*>* preds);

    Status get_key_ranges(std::vector<std::unique_ptr<OlapScanRange>>* key_ranges) {
        Status st = scan_keys.get_key_range(key_ranges);
        if (!st.ok()) return st;
        if (key_ranges->empty()) {
            key_ranges->emplace_back(new OlapScanRange());
        }
        return st;
    }

    void get_not_push_down_conjuncts(std::vector<ExprContext*>* predicates) {
        DCHECK_EQ(conjunct_ctxs_ptr->size(), normalized_conjuncts.size());
        for (size_t i = 0; i < normalized_conjuncts.size(); i++) {
            if (!normalized_conjuncts[i]) {
                predicates->push_back(conjunct_ctxs_ptr->at(i));
            }
        }
    }

    Status normalize_conjuncts();
    Status build_olap_filters();
    Status build_scan_keys(bool no_limit, int32_t max_scan_key_num);

private:
    template <PrimitiveType SlotType, typename RangeValueType>
    void normalize_predicate(const SlotDescriptor& slot, ColumnValueRange<RangeValueType>* range);

    template <PrimitiveType SlotType, typename RangeValueType>
    void normalize_in_or_equal_predicate(const SlotDescriptor& slot, ColumnValueRange<RangeValueType>* range);

    template <PrimitiveType SlotType, typename RangeValueType>
    void normalize_binary_predicate(const SlotDescriptor& slot, ColumnValueRange<RangeValueType>* range);

    template <PrimitiveType SlotType, typename RangeValueType>
    void normalize_join_runtime_filter(const SlotDescriptor& slot, ColumnValueRange<RangeValueType>* range);

    template <PrimitiveType SlotType, typename RangeValueType>
    void normalize_not_in_or_not_equal_predicate(const SlotDescriptor& slot, ColumnValueRange<RangeValueType>* range);

    void normalize_is_null_predicate(const SlotDescriptor& slot);
};

} // namespace vectorized
} // namespace starrocks